前言

寫到這篇不得不讚嘆 Gradio，在 Python 成為最簡單的程式語言的今天，竟然直到最近才有一個能同時撰寫簡單的程式碼就能同時架設前端與後端，真的不得不讚嘆這些開源開發者。

既然我們寫了 Embedding -> 文件前處理 -> 一個套件能同時處理前端與後端，再來我們來實作一個 demo 能運行這些實作吧！

前情提要

我們會使用：

• 向量儲存：Qdrant
• 文件處理：Marker
• LLM Runtime：Ollama
• 前後端處理：Gradio

這次的 demo 中，由於 Gemma3 的能力表現優於 llama3.2，所以這邊也先建議進行抓取模型

ollama pull gemma3:12b

並且定義全域的變數，方便後面使用

import re
from uuid import uuid4

import ollama
import gradio as gr
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct
from marker.models import create_model_dict
from marker.output import text_from_rendered
from marker.config.parser import ConfigParser
from marker.converters.pdf import PdfConverter

qdrant = QdrantClient(":memory:")
client = ollama.Client()

建立向量資料庫

def create_collection():
    """
    重新建立 collection
    """
    qdrant.recreate_collection(
        collection_name="docs",
        vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
    )

文字轉換向量

def get_embedding(text):
    """
    轉換向量
    """
    result = client.embeddings(
        model="bge-m3:567m", 
        prompt=text,
        keep_alive="1s"
    )["embedding"]
    
    return result

實作上傳檔案＆向量儲存

def process_file(file_path):
    """
    將指定路徑的檔案轉換成向量並請上傳到向量資料庫
    """
    converter = PdfConverter(
        config=ConfigParser({
            "paginate_output": True
        }).generate_config_dict(),
        artifact_dict=create_model_dict()
    )
    rendered = converter(file_path)
    text_list, _, _ = text_from_rendered(rendered)
    text_list = re.split(r"\{\d+\}-+", text_list)
    text_list.pop(0) # pop(0) 是因為 ["", "<page_1>", "<page_2>",...]

    data_list = []
    for text in text_list:
        data_list.append({
            "embedding": get_embedding(text),
            "metadata": {
                "source_text": text
            }
        })

        qdrant.upsert(
            collection_name="docs",
            points=[
                PointStruct(
                    id=str(uuid4()),
                    vector=get_embedding(text),
                    payload={"text": text}
                )
            ]
        )

這邊一樣因為我們需要在搜尋出向量資料時將原文件的內容復原到我們的 Prompt 中，所以我們這邊比照先前的做法，在每筆向量資料中加上原始文件的內容

定義 Gradio UI 及元件監聽器行為

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column() as file:
            upload = gr.File()
            
        with gr.Column() as chatbot:
            chatbot = gr.Chatbot(type="messages")
            
            with gr.Group():
                textbox = gr.Textbox()
                submit = gr.Button()
    
    upload.upload(
        process_file,
        inputs=[upload],
        outputs=[upload]
    )
    submit.click(
        send_task,
        inputs=[textbox, chatbot],
        outputs=[textbox, chatbot]
    )

定義行為函數

def process_file(file_path):
    """
    將指定路徑的檔案轉換成向量並請上傳到向量資料庫
    """
    converter = PdfConverter(
        config=ConfigParser({
            "paginate_output": True
        }).generate_config_dict(),
        artifact_dict=create_model_dict()
    )
    rendered = converter(file_path)
    text_list, _, _ = text_from_rendered(rendered)
    text_list = re.split(r"\{\d+\}-+", text_list)
    text_list.pop(0)

    data_list = []
    for text in text_list:
        data_list.append({
            "embedding": get_embedding(text),
            "metadata": {
                "source_text": text
            }
        })

        qdrant.upsert(
            collection_name="docs",
            points=[
                PointStruct(
                    id=str(uuid4()),
                    vector=get_embedding(text),
                    payload={"text": text}
                )
            ]
        )

def send_task(query, history):
    """
    與 RAG LLM 互動
    """
    if query.strip() == "":
        yield gr.update(), gr.update()
        return
    
    history.append(
        {
            "role": "user",
            "content": query.strip()
        }
    )
    
    yield gr.update(value=""), gr.update(value=history)
    
    search_result = qdrant.search(
        collection_name="docs",
        query_vector=get_embedding(query.strip()),
        limit=2
    )
    
    if len(history) > 0 and history[0]["role"] != "system":
        history.insert(0, {
            "role": "system",
            "content": """
            請根據以下的提示進行回覆
            
            {}
            """.format('\n'.join(i.payload['text'] for i in search_result))
        })
    
    else:
        history[0] = {
            "role": "system",
            "content": """
            請根據以下的提示進行回覆
            
            {}
            """.format('\n'.join(i.payload['text'] for i in search_result))
        }
    
    result = client.chat(
        model="gemma3:12b",
        messages=[{"role": i["role"], "content": i["content"]} for i in history]
    ).message.content
    
    history.append(
        {
            "role": "assistant",
            "content": result
        }
    )
    
    yield gr.update(), gr.update(value=history)
    return

程式碼解釋

line 59: 這邊的判斷式是因為一個 history 裡面最好只有一個 「System Prompt」，所以就是有 System prompt 就覆寫 RAG 回覆，沒有就插入一個訊息

line 81: 這邊這個做法可以做到擁有 「歷史記憶」

最後成果

• 關於內輪差的敘述，何者錯誤?

  

• 騎機車時若發現煞車系統失靈，第一時間應如何作較為適當?